﻿using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;
using Confluent.Kafka;
using static System.Windows.Forms.VisualStyles.VisualStyleElement.ToolTip;
using Google.Protobuf;

namespace Yunwei
{
    public partial class KafkaSendForm : Form
    {
        private IProducer<string, byte[]> producer; // Change TValue to byte[]  
        private bool isProducerInitialized = false;
        public KafkaSendForm()
        {
            InitializeComponent();
        }

        private async void buttonSend_ClickAsync(object sender, EventArgs e)
        {
            string BootstrapServers = textBoxKafkaAddr.Text;
            string Paraid = textParaid.Text;

            string CollectTimeStr = textCollectTime.Text;
            long CollectTime = ConvertTimeStringToLong(CollectTimeStr);
            float paraValue = float.Parse(textBoxParaValue.Text);


            Standard standard1 = new Standard
            {
                Paraid = Paraid,
                Value = paraValue,
                CollectTime = CollectTime,
                DataType = "report"
            };

            Compose compose = new Compose
            {
                ContentType = "oridata"
            };

            compose.Standards.Add(standard1);

            MessageBox.Show($"ip:{BootstrapServers},rtuCode: {Paraid}, collectTime: {CollectTime}, paraValue: {paraValue}");

            byte[] bytes = compose.ToByteArray();

            if (!isProducerInitialized)
            {
                var config = new ProducerConfig { BootstrapServers = BootstrapServers };
                producer = new ProducerBuilder<string, byte[]>(config).Build(); // Ensure TValue matches byte[]  
                isProducerInitialized = true;
            }
            try
            {
                var result = await producer.ProduceAsync("distribute", new Message<string, byte[]> { Key = "data", Value = bytes });
                MessageBox.Show($"消息已发送到分区 {result.Partition}，偏移量 {result.Offset}。");
            }
            catch (ProduceException<Null, byte[]> ex) // Update exception type to match TValue  
            {
                MessageBox.Show($"发送消息时出错: {ex.Error.Reason}");
            }
        }


        static long ConvertTimeStringToLong(string timeString)
        {
            // 定义时间字符串的格式
            string format = "yyyy-MM-dd HH:mm:ss";
            // 尝试将时间字符串转换为 DateTime 类型
            if (DateTime.TryParseExact(timeString, format, null, System.Globalization.DateTimeStyles.None, out DateTime dateTime))
            {
                // 获取 Unix 时间戳的起始时间
                DateTime unixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
                // 将 DateTime 转换为 UTC 时间
                DateTime utcDateTime = dateTime.ToUniversalTime();
                // 计算时间差并转换为毫秒
                return (long)(utcDateTime - unixEpoch).TotalMilliseconds;
            }
            else
            {
                // 若转换失败，输出错误信息并返回 0
                Console.WriteLine("时间字符串格式不正确。");
                return 0;
            }
        }
    }
}
